package saic.spark_stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaUtils
import kafka.serializer.StringDecoder

/**
 * @author ZhiLi
 * The code from net
 */
object UserClickCountAnalytics  {
  def main(args: Array[String]): Unit = {
    var masterUrl = "local[1]"
    if (args.length > 0) {
      masterUrl = args(0)
    }
 
    // Create a StreamingContext with the given master URL
    val conf = new SparkConf().setMaster(masterUrl).setAppName("UserClickCountStat")
    val ssc = new StreamingContext(conf, Seconds(5))
 
    // Kafka configurations
    val topics = Set("mykafka01")//kafka topics set
//    val brokers = "172.16.85.129:9092"//kafka broker list
    val brokers = "10.122.17.129:9095,10.122.17.130:9095,10.122.17.131:9095"
    //kafka params
    val kafkaParams = Map[String, String](
        //"zoo"   -> "",
      "metadata.broker.list" -> brokers, 
      "serializer.class" -> "kafka.serializer.StringEncoder"//kafka serializer classs
      )
    
 
    // Create a direct stream
    val kafkaStreamRDDs = KafkaUtils.createDirectStream[String, String,
      StringDecoder, StringDecoder](ssc, kafkaParams, topics)
      
 
    kafkaStreamRDDs.foreachRDD(rdd => 
      rdd.foreach(line => 
        print(line._1+"+++"+line._2)
        )
    )
 
 
    ssc.start()
    ssc.awaitTermination()
 
  }
}